package com.wsoft.utils;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;

/**
 * @Author z001
 * @Date 2025/1/8 14:18
 */
@Component
public class RabbitMQUtil {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private AmqpAdmin amqpAdmin;

    private Connection connection;
    private Channel channel;

    @Value("${spring.rabbitmq.host}")
    private String host;
    @Value("${spring.rabbitmq.port}")
    private Integer port;
    @Value("${spring.rabbitmq.username}")
    private String userName;
    @Value("${spring.rabbitmq.password}")
    private String passWord;

    /**
     * 初始化连接和通道
     */
    @PostConstruct
    public void init() throws Exception {
        // 创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(host);
        factory.setPort(port);
        factory.setUsername(userName);
        factory.setPassword(passWord);

        // 创建连接和通道
        connection = factory.newConnection();
        channel = connection.createChannel();
    }

    /**
     * 发送消息到指定交换机
     *
     * @param exchange   交换机名称
     * @param routingKey 路由键
     * @param message    消息内容
     */
    public void sendMessage(String exchange, String routingKey, Object message) {
        rabbitTemplate.convertAndSend(exchange, routingKey, message);
        System.out.println(" [x] Sent to exchange '" + exchange + "' with routing key '" + routingKey + "': " + message);
    }

    /**
     * 手动应答消息
     * @param deliveryTag
     * @param multiple
     *  false : 确认当前 deliveryTag 对应的单条消息。
     *  true: 确认当前 deliveryTag 及其之前的所有未确认消息。
     * @throws Exception
     */
    public void manualAck(long deliveryTag,boolean multiple) throws Exception {
        channel.basicAck(deliveryTag, multiple);
        System.out.println(" [x] Acknowledged message with delivery tag: " + deliveryTag);
    }

    /**
     * 拒绝单条消息（basicReject）
     * @param deliveryTag
     * @param requeue
     *  requeue = true：消息重新放回队列，适合需要重试的场景。
     *  requeue = false：消息被丢弃或进入死信队列，适合无法处理的消息。
     * @throws Exception
     */
    public void rejectMessage(long deliveryTag, boolean requeue) throws Exception {
        channel.basicReject(deliveryTag, requeue);
        System.out.println(" [x] Rejected message with delivery tag: " + deliveryTag + ", requeue: " + requeue);
    }

    /**
     * 发送延迟消息到指定交换机
     *
     * @param exchange   交换机名称
     * @param routingKey 路由键
     * @param message    消息内容
     * @param delay      延迟时间（单位：毫秒）
     */
    public void sendDelayedMessage(String exchange, String routingKey, String message, int delay) {
        rabbitTemplate.convertAndSend(exchange, routingKey, message, msg -> {
            msg.getMessageProperties().setHeader("x-delay", delay); // 设置延迟时间
            System.out.println(" [x] Sent to exchange '" + exchange + "' with routing key '" + routingKey + "': " + message + ", Delay: " + delay + "ms");
            return msg;
        });
    }

    /**
     * 创建直连交换机
     *
     * @param exchangeName 交换机名称
     * @param durable      是否持久化
     * @param autoDelete   是否自动删除
     */
    public void createDirectExchange(String exchangeName, boolean durable, boolean autoDelete) {
        DirectExchange exchange = new DirectExchange(exchangeName, durable, autoDelete);
        amqpAdmin.declareExchange(exchange);
        System.out.println(" [x] Created direct exchange: " + exchangeName);
    }

    /**
     * 创建主题交换机
     *
     * @param exchangeName 交换机名称
     * @param durable      是否持久化
     * @param autoDelete   是否自动删除
     */
    public void createTopicExchange(String exchangeName, boolean durable, boolean autoDelete) {
        TopicExchange exchange = new TopicExchange(exchangeName, durable, autoDelete);
        amqpAdmin.declareExchange(exchange);
        System.out.println(" [x] Created topic exchange: " + exchangeName);
    }

    /**
     * 创建延迟交换机
     *
     * @param exchangeName 交换机名称
     * @param durable      是否持久化
     * @param autoDelete   是否自动删除
     */
    public void createDelayedExchange(String exchangeName, boolean durable, boolean autoDelete) {
        Map<String, Object> arguments = new HashMap<>(1);
        // 指定延迟交换机的类型
        arguments.put("x-delayed-type", "direct");
        CustomExchange exchange = new CustomExchange(exchangeName, "x-delayed-message", durable, autoDelete, arguments);
        amqpAdmin.declareExchange(exchange);
        System.out.println(" [x] Created delayed exchange: " + exchangeName);
    }

    /**
     * 创建队列
     *
     * @param queueName 队列名称
     * @param durable   是否持久化
     * @param exclusive 是否独占
     * @param autoDelete 是否自动删除
     */
    public void createQueue(String queueName, boolean durable, boolean exclusive, boolean autoDelete) {
        Queue queue = new Queue(queueName, durable, exclusive, autoDelete);
        amqpAdmin.declareQueue(queue);
        System.out.println(" [x] Created queue: " + queueName);
    }

    /**
     * 绑定队列到交换机
     *
     * @param queueName    队列名称
     * @param exchangeName 交换机名称
     * @param routingKey   路由键
     */
    public void bindQueueToExchange(String queueName, String exchangeName, String routingKey) {
        Binding binding = new Binding(queueName, Binding.DestinationType.QUEUE, exchangeName, routingKey, null);
        amqpAdmin.declareBinding(binding);
        System.out.println(" [x] Bound queue '" + queueName + "' to exchange '" + exchangeName + "' with routing key '" + routingKey + "'");
    }

    /**
     * 绑定队列到延迟交换机
     *
     * @param queueName    队列名称
     * @param exchangeName 交换机名称
     * @param routingKey   路由键
     */
    public void bindQueueToDelayedExchange(String queueName, String exchangeName, String routingKey) {
        Binding binding = BindingBuilder.bind(new Queue(queueName))
                .to(new CustomExchange(exchangeName, "x-delayed-message", true, false))
                .with(routingKey)
                .noargs();
        amqpAdmin.declareBinding(binding);
        System.out.println(" [x] Bound queue '" + queueName + "' to delayed exchange '" + exchangeName + "' with routing key '" + routingKey + "'");
    }

    /**
     * 删除队列
     *
     * @param queueName 队列名称
     */
    public void deleteQueue(String queueName) {
        amqpAdmin.deleteQueue(queueName);
        System.out.println(" [x] Deleted queue: " + queueName);
    }

    /**
     * 删除交换机
     *
     * @param exchangeName 交换机名称
     */
    public void deleteExchange(String exchangeName) {
        amqpAdmin.deleteExchange(exchangeName);
        System.out.println(" [x] Deleted exchange: " + exchangeName);
    }
}
